package com.whoami.dataplatform.tag_combine.old;

import org.apache.commons.lang3.StringUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.BytesWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.LazyOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.MultipleOutputs;
import org.apache.hadoop.mapreduce.lib.output.SequenceFileAsBinaryOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;
import org.apache.orc.mapred.OrcList;
import org.apache.orc.mapred.OrcStruct;
import org.apache.orc.mapreduce.OrcInputFormat;

import java.io.FileInputStream;
import java.io.IOException;
import java.util.BitSet;
import java.util.Properties;
import java.util.stream.Collectors;

/**
 * 给所有被打上标签的设备id编号, 方便下一步转为bitmap
 * <p>
 * 输入
 * /user/mz_supertool/datalake/src/segment/dwd_segment/20190723/data
 * 000Lh0rMaC79  [{"id":"23","name":"休闲娱乐"},{"id":"32","name":"视频/短视频"}]
 *
 * @author tzp
 * @since 2019/7/31
 */
@SuppressWarnings("Duplicates")
public class UIDToNumberJob extends Configured implements Tool {
    /**
     * 给MR框架的一堆配置, 文件里面的内容会被set到mr conf里
     */
    public static final String PARAMS_MR_PROP = "mr_properties";

    @Override
    public int run(String[] args) throws Exception {
        Configuration conf = getConf();
        String mrProFile = conf.get(PARAMS_MR_PROP);
        if (StringUtils.isNotEmpty(mrProFile)) {
            Properties mrProperties = new Properties();
            mrProperties.load(new FileInputStream(mrProFile));
            for (Object key : mrProperties.keySet()) {
                conf.set(key.toString(), mrProperties.get(key).toString());
            }
        }
        String date = args[0];//20190723
        String input = args[1];
        String output = args[2];//outputbase

        Job job = Job.getInstance(conf, this.getClass().getSimpleName() + " " + date);
        job.setJarByClass(this.getClass());

        //输入
        job.setInputFormatClass(OrcInputFormat.class);
        FileInputFormat.addInputPath(job, new Path(input));

        //mapper
        job.setMapperClass(SimpleMapper.class);
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(Text.class);

        //reducer
        job.setReducerClass(SortNumberReducer.class);
        job.setNumReduceTasks(64);

        //输出
        job.setOutputKeyClass(BytesWritable.class);
        job.setOutputValueClass(BytesWritable.class);
        job.setOutputFormatClass(LazyOutputFormat.class);
        LazyOutputFormat.setOutputFormatClass(job, SequenceFileAsBinaryOutputFormat.class);

        FileOutputFormat.setOutputPath(job, new Path(output));

        boolean result = job.waitForCompletion(true);

        return result ? 0 : 1;
    }

    public static class SimpleMapper extends Mapper<NullWritable, OrcStruct, Text, Text> {
        private Text outKey = new Text();
        private Text outValue = new Text();

        @Override
        protected void map(NullWritable key, OrcStruct value, Context context) throws IOException, InterruptedException {
            String uid = value.getFieldValue("uid").toString().trim().toUpperCase();
            outKey.set(uid);

            OrcList<OrcStruct> segments = (OrcList<OrcStruct>) value.getFieldValue("segment");
            String ids = segments.stream()
                    .map((x) -> x.getFieldValue("id").toString())
                    .collect(Collectors.joining(","));
            outValue.set(ids);
            context.write(outKey, outValue);
        }
    }

    /**
     * 输出:
     * uid -> tid_bits (usn暂时不用)
     * usn -> uid (方法一用)
     * tid -> usn (方法一用)
     * tid_bits -> uid (方法二用)
     * <p>
     * --------------- 下一个任务
     * <p>
     * tid -> usn_bits
     * <p>
     * tid_bits -> num, download?usns?
     */
    public static class SortNumberReducer extends Reducer<Text, Text, BytesWritable, BytesWritable> {
        BytesWritable outKey = new BytesWritable();
        BytesWritable outValue = new BytesWritable();
        private MultipleOutputs<BytesWritable, BytesWritable> mos;

        Long number = 0L;
        BitSet tidbits = new BitSet(1024);

        @Override
        protected void setup(Context context) throws IOException, InterruptedException {
            mos = new MultipleOutputs<>(context);
        }

        @Override
        protected void reduce(Text key, Iterable<Text> values, Context context) throws IOException, InterruptedException {
            tidbits.clear();
            byte[] uidBytes = key.getBytes();
//            byte[] usnBytes = longtoBytes(number);
            //usn -> uid
//            outKey.set(usnBytes, 0, usnBytes.length);
//            outValue.set(uidBytes, 0, key.getLength());
//            mos.write(outKey, outValue, "usn-uid");

            for (Text value : values) {
                for (String tid : value.toString().split(",")) {
                    tidbits.set(Integer.parseInt(tid));
                }
            }
            byte[] tidBytes = tidbits.toByteArray();

            //uid -> tid_bits
//            outKey.set(uidBytes, 0, key.getLength());
//            outValue.set(tidBytes, 0, tidBytes.length);
//            mos.write(outKey, outValue, "uid-tid_bits");


            //tid -> usn
//            for (int i = tidbits.nextSetBit(0); i >= 0; i = tidbits.nextSetBit(i + 1)) {
//                byte[] bytes = longtoBytes(i);
//                outKey.set(bytes, 0, bytes.length);
//                outValue.set(usnBytes, 0, usnBytes.length);
//                mos.write(outKey, outValue, "tid-usn");
//            }

            //tid_bits -> uid
            outKey.set(tidBytes, 0, tidBytes.length);
            outValue.set(uidBytes, 0, key.getLength());
            mos.write(outKey, outValue, "tid_bits-uid");

            number++;
        }

        @Override
        protected void cleanup(Context context) throws IOException, InterruptedException {
            mos.close();
        }
    }

    private static byte[] longtoBytes(long data) {
        return new byte[]{
                (byte) ((data >> 56) & 0xff),
                (byte) ((data >> 48) & 0xff),
                (byte) ((data >> 40) & 0xff),
                (byte) ((data >> 32) & 0xff),
                (byte) ((data >> 24) & 0xff),
                (byte) ((data >> 16) & 0xff),
                (byte) ((data >> 8) & 0xff),
                (byte) ((data >> 0) & 0xff),
        };
    }

    public static void main(String[] args) throws Exception {
        System.exit(ToolRunner.run(new Configuration(), new UIDToNumberJob(), args));
    }
}
